package com.yanboo.modules.mnt.websocket.processor;

import com.alibaba.fastjson.JSON;
import com.yanboo.common.config.UidGenerator;
import com.yanboo.common.utils.SpringContextHolder;
import com.yanboo.constant.ProcessorMsgType;
import com.yanboo.modules.mnt.domain.Processor;
import com.yanboo.modules.mnt.monitor.MonitorContext;
import com.yanboo.modules.mnt.repository.ProcessorRepository;
import com.yanboo.modules.mnt.service.ProcessorService;
import com.yanboo.modules.mnt.service.dto.ProcessorDto;
import com.yanboo.modules.mnt.service.dto.ProcessorQueryCriteria;
import com.yanboo.modules.mnt.websocket.processor.dto.ProcessorWSMsgDto;
import jakarta.websocket.*;
import jakarta.websocket.server.PathParam;
import jakarta.websocket.server.ServerEndpoint;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.logging.log4j.util.Strings;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.IdGenerator;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

@Slf4j
@Component
@ServerEndpoint("/ws/processor/{id}")
public class LogWSService {

    static ConcurrentHashMap<String, List<Session>> sessionConcurrentHashMap = new ConcurrentHashMap<>();

    static ConcurrentHashMap<String, List<Session>> pathConcurrentHashMap = new ConcurrentHashMap<>();


    @OnOpen
    public void onOpen(@PathParam("id") String id, Session session) throws IOException {
        if (Strings.isEmpty(id)) {
            session.close();
        }
        ProcessorService processorService = SpringContextHolder.getBean(ProcessorService.class);
        ProcessorDto processorDto = processorService.findById(Long.parseLong(id));
        String logPath = null;
        if (processorDto.getCommand().contains(">")) {
            String s = processorDto.getCommand().split(">")[1].trim();
            if (s.contains(" ")) {
                logPath = s.split(" ")[0];
            } else {
                logPath = s;
            }
        } else {
            logPath = System.getProperty("user.dir") + "/logs/" + processorDto.getId() + ".log";
        }
        File file = new File(logPath);
        if (!file.exists()) {
            throw new IOException("文件不存在");
        }
        if (pathConcurrentHashMap.containsKey(file.getPath())) {
            pathConcurrentHashMap.get(file.getPath()).add(session);
        } else {
            ArrayList<Session> value = new ArrayList<>();
            value.add(session);
            pathConcurrentHashMap.put(file.getPath(), value);
        }

        if (!MonitorContext.isRegistered(file.getPath())) {
            MonitorContext.register(file.getPath(), LogWSService::sendMsg);
        }
        List<Session> sessions = new ArrayList<>();
        if (sessionConcurrentHashMap.containsKey(id)) {
            sessions = sessionConcurrentHashMap.get(id);
            sessions.add(session);
            return;
        }
        sessions.add(session);
        sessionConcurrentHashMap.put(id, sessions);
    }

    @OnClose
    public void onClose(CloseReason closeReason, Session session) {
        log.info("[websocket] 连接断开：id={}，reason={}", session.getId(), closeReason.getReasonPhrase());
        sessionConcurrentHashMap.forEach((k, v) -> {
            Optional<Session> optSession = v.stream().filter(x -> x.getId().equals(session.getId())).findFirst();
            optSession.ifPresent(v::remove);
        });
        pathConcurrentHashMap.forEach((k, v) -> {
            Optional<Session> optSession = v.stream().filter(x -> x.getId().equals(session.getId())).findFirst();
            optSession.ifPresent(v::remove);
            if (v.isEmpty()){
                MonitorContext.unRegister(k);
            }
        });
    }


    // 连接异常
    @OnError
    public void onError(Throwable throwable, Session session) throws IOException {
        log.info("[websocket] 连接异常：id={}，throwable={}", session.getId(), throwable.getMessage());
        sessionConcurrentHashMap.forEach((k, v) -> {
            Optional<Session> optSession = v.stream().filter(x -> x.getId().equals(session.getId())).findFirst();
            optSession.ifPresent(v::remove);
        });
        pathConcurrentHashMap.forEach((k, v) -> {
            Optional<Session> optSession = v.stream().filter(x -> x.getId().equals(session.getId())).findFirst();
            optSession.ifPresent(v::remove);
            if (v.isEmpty()){
                MonitorContext.unRegister(k);
            }
        });
        // 关闭连接。状态码为 UNEXPECTED_CONDITION（意料之外的异常）
        session.close(new CloseReason(CloseReason.CloseCodes.UNEXPECTED_CONDITION, throwable.getMessage()));
    }


    public static void sendMsg(String path, String line) {
        ProcessorWSMsgDto processorWSMsgDto = new ProcessorWSMsgDto().setId(UUID.randomUUID().toString()).setType(ProcessorMsgType.LOG).setMsg(line);
        sendMsg(path, processorWSMsgDto);
    }

    private static void sendMsg(String path, ProcessorWSMsgDto message) {
        List<Session> sessions = pathConcurrentHashMap.get(path);
        if (sessions == null || sessions.isEmpty()) {
            return;
        }
        String jsonString = JSON.toJSONString(message);
        sessions.forEach(x -> {
            if (x.isOpen()) {
                x.getAsyncRemote().sendText(jsonString);
            }
        });
        log.info("websocket send message:" + jsonString);
    }

    public static void sendMsg(Long id, ProcessorWSMsgDto message) {
        List<Session> sessions = sessionConcurrentHashMap.get(id.toString());
        if (sessions == null || sessions.isEmpty()) {
            return;
        }
        String jsonString = JSON.toJSONString(message);
        sessions.forEach(x -> {
            x.getAsyncRemote().sendText(jsonString);
        });
        log.info("websocket send message:" + jsonString);
    }

}
